/*
 * @Author: gonglf
 * @Date: 2022-07-08 15:37:24
 * @LastEditors: gonglf
 * @LastEditTime: 2022-07-08 16:55:59
 * @Description:服务管理
 *
 */
package asynctask

import (
	"errors"
	"os"
	"os/signal"
	"sync"
	"syscall"

	"github.com/e421083458/golang_common/log"
)

type Worker struct {
	server            *Server
	ConsumerTag       string
	Concurrency       int
	Queue             string
	errorHandler      func(err error)
	preTaskHandler    func(*tasks.Signature)
	postTaskHandler   func(*tasks.Signature)
	preConsumeHandler func(*Worker) bool
}

var (
	// ErrWorkerQuitGracefully is return when worker quit gracefully
	ErrWorkerQuitGracefully = errors.New("Worker quit gracefully")
	// ErrWorkerQuitGracefully is return when worker quit abruptly
	ErrWorkerQuitAbruptly = errors.New("Worker quit abruptly")
)

//启动一个新的工作进程，worker订阅到默认队列并处理注册的任务
func (worker *Worker) Launch() error {
	errorsChan := make(chan error)

	worker.LaunchAsync(errorsChan)

	return <-errorsChan
}

func (worker *Worker) LaunchAsync(errorsChan chan<- error) {
	cnf := worker.server.GetConfig()
	broker := worker.server.GetBroker()

	// Log some useful information about worker configuration
	log.INFO.Printf("Launching a worker with the following settings:")
	// log.INFO.Printf("- Broker: %s", RedactURL(cnf.Broker))
	if worker.Queue == "" {
		log.INFO.Printf("- DefaultQueue: %s", cnf.DefaultQueue)
	} else {
		log.INFO.Printf("- CustomQueue: %s", worker.Queue)
	}
	// log.INFO.Printf("- ResultBackend: %s", RedactURL(cnf.ResultBackend))
	// if cnf.AMQP != nil { //这个是存放结果的
	// 	log.INFO.Printf("- AMQP: %s", cnf.AMQP.Exchange)
	// 	log.INFO.Printf("  - Exchange: %s", cnf.AMQP.Exchange)
	// 	log.INFO.Printf("  - ExchangeType: %s", cnf.AMQP.ExchangeType)
	// 	log.INFO.Printf("  - BindingKey: %s", cnf.AMQP.BindingKey)
	// 	log.INFO.Printf("  - PrefetchCount: %d", cnf.AMQP.PrefetchCount)
	// }

	var signalWG sync.WaitGroup
	// 开启消费broker和broker断开重连
	go func() {
		for {
			retry, err := broker.StartConsuming(worker.ConsumerTag, worker.Concurrency, worker) //这边会被阻塞

			if retry {
				if worker.errorHandler != nil {
					worker.errorHandler(err)
				} else {
					log.WARNING.Printf("Broker failed with error: %s", err)
				}
			} else {
				signalWG.Wait()
				errorsChan <- err // stop the goroutine
				return
			}
		}
	}()
	//处理退出信号
	if !cnf.NoUnixSignals {
		sig := make(chan os.Signal, 1)
		signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
		var signalsReceived uint

		// Goroutine Handle SIGINT and SIGTERM signals
		go func() {
			for s := range sig { //这边会一直阻塞直到有信号
				log.WARNING.Printf("Signal received: %v", s)
				signalsReceived++

				if signalsReceived < 2 {
					// After first Ctrl+C start quitting the worker gracefully
					log.WARNING.Print("Waiting for running tasks to finish before shutting down")
					signalWG.Add(1)
					go func() {
						worker.Quit() //退出
						errorsChan <- ErrWorkerQuitGracefully
						signalWG.Done()
					}()
				} else {
					// Abort the program when user hits Ctrl+C second time in a row
					errorsChan <- ErrWorkerQuitAbruptly
				}
			}
		}()
	}
}

//退出正在运行的worker进程
func (worker *Worker) Quit() {
	worker.server.GetBroker().StopConsuming()
}

//自定义的任务队列
func (worker *Worker) CustomQueue() string {
	return worker.Queue
}

//接收任务处理，call func
func (worker *Worker) Process(signature *tasks.Signature) error {
	//判断任务是否存在
	if !worker.server.IsTaskRegistered(signature.Name) {
		return nil
	}

	taskFunc, err := worker.server.GetRegisteredTask(signature.Name)
	if err != nil {
		return nil
	}

	// Update task state to RECEIVED
	// if err = worker.server.GetBackend().SetStateReceived(signature); err != nil {
	// 	return fmt.Errorf("Set state to 'received' for task %s returned error: %s", signature.UUID, err)
	// }

	//准备要处理的任务
	task, err := tasks.NewWithSignature(taskFunc, signature)
	// if this failed, it means the task is malformed, probably has invalid
	// signature, go directly to task failed without checking whether to retry
	if err != nil {
		worker.taskFailed(signature, err)
		return err
	}

	// try to extract trace span from headers and add it to the function context
	// so it can be used inside the function if it has context.Context as the first
	// argument. Start a new span if it isn't found.
	// taskSpan := tracing.StartSpanFromHeaders(signature.Headers, signature.Name)
	// tracing.AnnotateSpanWithSignatureInfo(taskSpan, signature)
	// task.Context = opentracing.ContextWithSpan(task.Context, taskSpan)

	// Update task state to STARTED
	// if err = worker.server.GetBackend().SetStateStarted(signature); err != nil {
	// 	return fmt.Errorf("Set state to 'started' for task %s returned error: %s", signature.UUID, err)
	// }

	//Run handler before the task is called
	// if worker.preTaskHandler != nil {
	// 	worker.preTaskHandler(signature)
	// }

	// //Defer run handler for the end of the task
	// if worker.postTaskHandler != nil {
	// 	defer worker.postTaskHandler(signature)
	// }

	//调用任务
	results, err := task.Call()
	if err != nil {
		// If a tasks.ErrRetryTaskLater was returned from the task,
		// retry the task after specified duration
		retriableErr, ok := interface{}(err).(tasks.ErrRetryTaskLater)
		if ok {
			return worker.retryTaskIn(signature, retriableErr.RetryIn())
		}

		// Otherwise, execute default retry logic based on signature.RetryCount
		// and signature.RetryTimeout values
		if signature.RetryCount > 0 {
			return worker.taskRetry(signature)
		}

		return worker.taskFailed(signature, err)
	}

	return worker.taskSucceeded(signature, results)
}
